package com.bbx.flink.case1;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.arch.Processor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;

/**
 * 欺诈检测器
 */
@Slf4j
public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {
    private static final long serialVersionUID = 1L;

    private static final double SMALL_AMOUNT = 1.00;
    private static final double LARGE_AMOUNT = 500.00;
    private static final long ONE_MINUTE =  60*1000;

    private transient ValueState<Boolean> flagState;
    private transient ValueState<Long> timerState;


    @Override
    public void processElement(
            Transaction transaction,
            Context context,
            Collector<Alert> collector) throws Exception{
        //获取当前key 的当前flag
        Boolean lastTransactionWasSmall = flagState.value();
        if (lastTransactionWasSmall !=null) {
            if (transaction.getAmount()>LARGE_AMOUNT) {
                Alert alert = new Alert();
                alert.setId(transaction.getAccountId());
                collector.collect(alert);
                log.info("进入业务处理方法,AccountId[{}],Amount[{}],Timestamp[{}]",transaction.getAccountId(),
                        transaction.getAmount(),transaction.getTimestamp());
            }
            //清除当前state
            cleanUp(context);
//            flagState.clear();
        }
        if (transaction.getAmount()<SMALL_AMOUNT) {
            //设置状态
            flagState.update(true);
            //设置时间
            long timer = context.timerService().currentProcessingTime() + ONE_MINUTE;
            context.timerService().registerProcessingTimeTimer(timer);
            timerState.update(timer);
        }


    }

    /**
     * 设置状态
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        //状态
        ValueStateDescriptor<Boolean> flagDescriptor = new ValueStateDescriptor<Boolean>("Flag", Types.BOOLEAN);
         flagState = getRuntimeContext().getState(flagDescriptor);
         log.info("[{}]","进入open 方法");
         //时间
        ValueStateDescriptor<Long> timeDescriptor = new ValueStateDescriptor<>("time-state", Types.LONG);
        timerState = getRuntimeContext().getState(timeDescriptor);
    }

    /**
     * 计数器触发时，重置标志
     * @param timestamp
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out) throws Exception {
//        log.info("[{}]","进入onTimer 方法");
         timerState.clear();
         flagState.clear();
    }



    private void cleanUp(Context ctx) throws Exception {
        // delete timer
        Long timer = timerState.value();
        ctx.timerService().deleteProcessingTimeTimer(timer);

        // clean up all state
        timerState.clear();
        flagState.clear();
    }





}
